package com.grf.hadoop.practice;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import java.io.IOException;
import java.util.Iterator;
import java.util.TreeMap;

/**
 * 气温 Mapper类
 */
public class TempMapper extends Mapper<LongWritable, Text, TempBean, NullWritable> {

    private TempBean k = new TempBean();
    private TreeMap<TempBean, NullWritable> tempMap = new TreeMap<>();

    @Override
    protected void map(LongWritable key, Text value, Context context)
            throws IOException, InterruptedException {
        String line = value.toString();
        String[] fields = line.split(" ");
        k.setTemp(Double.parseDouble(fields[1]));
        k.setYear(fields[0]);
        //向TreeMap添加数据
        tempMap.put(k,NullWritable.get());
        if(tempMap.size()>10) {
            tempMap.remove(tempMap.lastKey());
        }
    }

    @Override
    protected void cleanup(Context context)
            throws IOException, InterruptedException {
        Iterator<TempBean> bean =  tempMap.keySet().iterator();
        while (bean.hasNext()) {
            TempBean k = bean.next();
            context.write(k, NullWritable.get());
        }

    }
}
